package com.chief.watermark;

import org.apache.commons.lang3.time.DateUtils;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.util.Collector;

import java.time.Duration;

public class ProcessFunctionTest {

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        DataStreamSource<String> socketTextStream = env.socketTextStream("hadoop55", 7777);

        DataStream<Tuple3<String, Long, Integer>> tuple3SingleOutputStreamOperator = socketTextStream.map(new MapFunction<String, Tuple3<String, Long, Integer>>() {
            @Override
            public Tuple3<String, Long, Integer> map(String value) throws Exception {
                String[] split = value.split(",");
                return new Tuple3<>(split[0], DateUtils.parseDate(split[1], "yyyy-MM-dd HH:mm:ss").getTime(), Integer.valueOf(split[2]));
            }
        });

        ProcessFunction<Tuple3<String, Long, Integer>, Tuple3<String, Long, Integer>> processFunction = new ProcessFunction<Tuple3<String, Long, Integer>, Tuple3<String, Long, Integer>>() {
            @Override
            public void processElement(Tuple3<String, Long, Integer> value, Context ctx, Collector<Tuple3<String, Long, Integer>> out) throws Exception {
                System.out.println(value + " currentWatermark :" + ctx.timerService().currentWatermark());
                out.collect(value);
            }
        };

        SingleOutputStreamOperator<Tuple3<String, Long, Integer>> tuple3SingleOutputStreamOperator1 =
                tuple3SingleOutputStreamOperator.assignTimestampsAndWatermarks(
                        WatermarkStrategy.<Tuple3<String, Long, Integer>>forBoundedOutOfOrderness(Duration.ofMillis(0))
                                .withTimestampAssigner((e, t) -> e.f1))
                        .process(processFunction);

        tuple3SingleOutputStreamOperator1.keyBy(t -> t.f0)
                .window(TumblingEventTimeWindows.of(Time.seconds(5)))
                .min(2).print();

        env.execute();
    }

}
